import logging

from coin.util.queue.config import KafkaConfig
from coin.util.queue.reader import AsyncKafkaReader


class BaseSubscriber(object):
  def __init__(self, *, topics, kafka_config):
    self.kafka_config = kafka_config
    self._topics = topics
    self._reader = None
    self._shutdown_requested = False

  async def loop(self):
    self._reader = AsyncKafkaReader(self._topics, kafka_config=self.kafka_config)
    await self._reader.open()

    while not self._shutdown_requested:
      try:
        records = await self._reader.read(timeout_ms=100)
      except TimeoutError:
        continue

      try:
        self.on_kafka_records(records)
      except Exception as e:
        logging.exception('Exception raised from engine')
        self.on_exception(e)

  def on_kafka_records(self, records):
    for record in records:
      try:
        self.on_message(record.topic, record.value)
      except Exception as e:
        logging.exception('Exception raised from engine')
        self.on_exception(e)

  def on_message(self, topic, message):
    raise NotImplementError()

  def on_exception(self, exception):
    pass
